package com.dynamicthreadpool.spi.support;

import com.dynamicthreadpool.spi.ExtensionLoader;
import com.dynamicthreadpool.spi.QueueFactory;
import com.dynamicthreadpool.spi.ThreadPool;
import com.dynamicthreadpool.spi.URL;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 贪婪线程池实现
 * 优先创建线程而不是将任务放入队列
 * 从Dubbo 2.7版本提取的完整实现
 */
public class EagerThreadPool implements ThreadPool {

    @Override
    public ThreadPoolExecutor getExecutor(URL url) {
        String name = url.getParameter("threadname", "DubboThreadPool");
        int cores = url.getParameter("corethreads", url.getParameter("threads", 200));
        int threads = url.getParameter("threads", 200);
        int queues = url.getParameter("queues", 0);
        int alive = url.getParameter("alive", 60000);
        
        // 对于EagerThreadPool，我们需要使用TaskQueue
        TaskQueue<Runnable> taskQueue = new TaskQueue<>(queues <= 0 ? 200 : queues);
        
        // 创建线程池
        EagerThreadPoolExecutor executor = new EagerThreadPoolExecutor(cores, threads, alive, TimeUnit.MILLISECONDS, taskQueue,
                new NamedInternalThreadFactory(name, true), new AbortPolicyWithReport(name, url));
        taskQueue.setExecutor(executor);
        return executor;
    }

    /**
     * 贪婪线程池执行器
     */
    public static class EagerThreadPoolExecutor extends ThreadPoolExecutor {

        private final AtomicInteger submittedTaskCount = new AtomicInteger(0);

        public EagerThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
                                       TaskQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) {
            super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
        }

        public int getSubmittedTaskCount() {
            return submittedTaskCount.get();
        }

        @Override
        protected void afterExecute(Runnable r, Throwable t) {
            submittedTaskCount.decrementAndGet();
        }

        @Override
        public void execute(Runnable command) {
            if (command == null) {
                throw new NullPointerException();
            }
            submittedTaskCount.incrementAndGet();
            try {
                super.execute(command);
            } catch (RejectedExecutionException rx) {
                // 如果被拒绝，尝试入队
                final TaskQueue queue = (TaskQueue) super.getQueue();
                try {
                    if (!queue.retryOffer(command, 0, TimeUnit.MILLISECONDS)) {
                        submittedTaskCount.decrementAndGet();
                        throw new RejectedExecutionException("Queue capacity is full.", rx);
                    }
                } catch (InterruptedException x) {
                    submittedTaskCount.decrementAndGet();
                    throw new RejectedExecutionException(x);
                }
            } catch (Throwable t) {
                submittedTaskCount.decrementAndGet();
                throw t;
            }
        }
    }

    /**
     * 任务队列，用于贪婪线程池
     */
    public static class TaskQueue<R extends Runnable> extends LinkedBlockingQueue<R> {

        private static final long serialVersionUID = -2635853580887179627L;

        private transient volatile EagerThreadPoolExecutor executor;

        public TaskQueue(int capacity) {
            super(capacity);
        }

        public void setExecutor(EagerThreadPoolExecutor exec) {
            executor = exec;
        }

        public boolean retryOffer(R o, long timeout, TimeUnit unit) throws InterruptedException {
            if (executor.isShutdown()) {
                throw new RejectedExecutionException("Executor is shutdown");
            }
            return super.offer(o, timeout, unit);
        }

        @Override
        public boolean offer(R runnable) {
            if (executor == null) {
                throw new RejectedExecutionException("The task queue does not have executor");
            }

            int currentPoolThreadSize = executor.getPoolSize();
            // 如果有空闲线程，直接入队，让线程处理
            if (executor.getSubmittedTaskCount() < currentPoolThreadSize) {
                return super.offer(runnable);
            }

            // 如果线程数小于最大线程数，返回false，让线程池创建新线程
            if (currentPoolThreadSize < executor.getMaximumPoolSize()) {
                return false;
            }

            // 如果线程数已达最大值，入队
            return super.offer(runnable);
        }

        private boolean isShutdown() {
            return executor.isShutdown();
        }
    }

    /**
     * 命名内部线程工厂
     */
    static class NamedInternalThreadFactory extends NamedThreadFactory {
        public NamedInternalThreadFactory(String prefix, boolean daemon) {
            super(prefix, daemon);
        }
    }
}